package com.zyw.concurrent.lock;

import com.zyw.cache.redis.client.RedisClient;
import com.zyw.cache.redis.utils.JedisUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * 锁观察者
 *
 * @author zhangyw
 * @date 16/3/31 21:32 10:35 14:23
 */
public class LockObserver extends AbstractLockObserver {

    protected static final Logger logger = LoggerFactory.getLogger(LockObserver.class);

    private final String schema;
    private RedisClient client;
    private Object mutex = new Object();
    private Map<String, LockListener> lockMap = new ConcurrentHashMap<String, LockListener>();
    private boolean stoped = false;
    private long interval = 500;
    private boolean terminated = false;
    private CountDownLatch doneSignal = new CountDownLatch(1);

    public LockObserver(String schema) {
        this.schema = schema;
        SystemExitListener.addTerminateListener(new ExitHandler() {
            public void run() {
                stoped = true;
                try {
                    doneSignal.await();
                } catch (InterruptedException e) {
                    logger.warn(e.getLocalizedMessage());
                }
            }
        });
    }


    public void addLockListener(String key, LockListener listener) {
        if (terminated) {
            listener.lockError();
            return;
        }
        synchronized (mutex) {
            lockMap.put(key, listener);
        }
    }

    public void removeLockListener(String key) {
        synchronized (mutex) {
            lockMap.remove(key);
        }
    }

    @Override
    public void run() {
        while (!terminated) {
            long p1 = System.currentTimeMillis();
            Map<String, LockListener> clone = new HashMap<String, LockListener>();
            synchronized (mutex) {
                clone.putAll(lockMap);
            }
            Set<String> keyset = clone.keySet();
            if (keyset.size() > 0) {
                for (String key : keyset) {
                    LockListener ll = clone.get(key);
                    try {
                        if (tryLock(key, ll.getExpire())) {
                            ll.lockAcquired();
                            removeLockListener(key);
                        }
                    } catch (Exception e) {
                        ll.lockError();
                        removeLockListener(key);
                    }
                }
            } else {
                if (stoped) {
                    terminated = true;
                    doneSignal.countDown();
                    return;
                }
            }
            try {
                long p2 = System.currentTimeMillis();
                long cost = p2 - p1;
                if (cost <= interval) {
                    Thread.sleep(interval - cost);
                } else {
                    Thread.sleep(interval * 2);
                }
            } catch (InterruptedException e) {
                logger.error(e.getLocalizedMessage());
            }
        }

    }


    /**
     * 超时时间单位为s!!!
     *
     * @param key
     * @param expireInSecond
     * @return
     */
    public boolean tryLock(final String key, final long expireInSecond) {
        if (terminated)
            return false;

        try {
            client = getResource();
            final long tt = System.currentTimeMillis();
            final long expire = expireInSecond * 1000;
            final Long ne = tt + expire;
            final String finalKey = formatLockKey(key);
            Long res = client.setnx(finalKey, String.valueOf(ne));
            if (new Long(1).equals(res)) {
                return true;
            } else {
                String ex = client.get(finalKey);
                if (ex == null || tt > Long.valueOf(ex)) {
                    String old = client.getSet(finalKey, String.valueOf(ne + 1));
                    if (old == null || (ex != null && ex.equals(old))) {
                        return true;
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        } finally {
            returnResource(client);
        }
        return false;
    }

    public void unLock(final String key) {
        try {
            client = getResource();
            client.del(formatLockKey(key));
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        } finally {
            returnResource(client);
        }

    }

    private String formatLockKey(final String key) {
        return schema + "#" + key;
    }

    /**
     * @return
     */
    private RedisClient getResource() {
        return JedisUtils.getResource().getResource();
    }

    /**
     * @param redisClient redis client
     */
    private void returnResource(RedisClient redisClient) {
        JedisUtils.getResource().returnResource(redisClient);
    }
}
